"""
Case Type   : fullsql
Case Name   : log_min_duration_statement参数值过大时,慢函数查询无结果
Create At   : 2022/01/29
Owner       : zhao-pengcheng-g
Description :
    1.检查参数，修改配置:enable_copy_server_files=on
    2.检查参数，修改配置:instr_unique_sql_count=100000
    3.创建用户，创建表
    4.导入数据
    5.设置log_min_duration_statement='24d',
    设置track_stmt_stat_level='l0,l2',执行query
    7.查询dbe_perf.get_slow_sql_by_timestamp函数
    8.查询dbe_perf.get_full_sql_by_timestamp函数
    9.环境清理
Expect      :
    1.修改参数成功
    2.修改参数成功
    3.创建成功
    4.导入成功
    5.设置成功，执行query成功
    6.无记录
    7.有记录
    8.环境清理成功
History     :
"""

import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common


class Fullsql(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        text = '---Opengauss_Function_Fullsql_Case0015 start---'
        self.logger.info(text)
        self.userNode = Node('PrimaryDbUser')
        self.rootNode = Node('PrimaryRoot')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.table_name_1 = 't_part_fullsql_0015_1'
        self.table_name_2 = 't_partsupp_fullsql_0015_2'
        self.table_name_3 = 't_supplier_fullsql_0015_3'
        self.user_name = 'u_fullsql_0015'
        self.table_path = os.path.join(r'/home',
                                       self.userNode.ssh_user, 'copy_tables')
        text = '---prestep1:修改机器umask---'
        self.logger.info(text)
        umask_cmd = f"umask 0022;"
        self.rootNode.sh(umask_cmd).result()
        text = '---step1:检查参数，修改配置:enable_copy_server_files=on,' \
               'expect:成功---'
        self.logger.info(text)
        self.config_copy = 'enable_copy_server_files'
        self.show_copy_cmd = f'show {self.config_copy};'
        check_copy = self.sh_primy.execut_db_sql(self.show_copy_cmd)
        self.logger.info(check_copy)
        self.copy_default = check_copy.splitlines()[-2].strip()
        if 'on' != self.copy_default:
            result = self.sh_primy.execute_gsguc('reload',
                                                 self.constant.GSGUC_SUCCESS_MSG,
                                                 f'{self.config_copy}=on')
            self.assertTrue(result, '参数修改失败')
        chk_copy_msg = self.sh_primy.execut_db_sql(self.show_copy_cmd)
        self.logger.info(chk_copy_msg)
        self.common.equal_sql_mdg(chk_copy_msg, self.config_copy, 'on',
                                  '(1 row)', flag='1')
        text = '---step2:检查参数，修改配置:instr_unique_sql_count=100000,' \
               'expect:成功---'
        self.logger.info(text)
        self.config_instr = 'instr_unique_sql_count'
        self.show_instr_cmd = f'show {self.config_instr};'
        check_instr = self.sh_primy.execut_db_sql(self.show_instr_cmd)
        self.logger.info(check_instr)
        self.instr_default = check_instr.splitlines()[-2].strip()
        if '100000' != self.instr_default:
            result = self.sh_primy.execute_gsguc('reload',
                                                 self.constant.GSGUC_SUCCESS_MSG,
                                                 f'{self.config_instr}=100000')
            self.assertTrue(result, '参数修改失败')
        chk_instr_msg = self.sh_primy.execut_db_sql(self.show_instr_cmd)
        self.logger.info(chk_instr_msg)
        self.common.equal_sql_mdg(chk_instr_msg, self.config_instr, '100000',
                                  '(1 row)', flag='1')

        text = '---step3:清理历史记录,创建用户，创建表,expect;成功---'
        self.logger.info(text)
        sql_cmd = f'''select reset_unique_sql('global', 'all', 1);
                    drop user if exists {self.user_name} cascade;
                    create user {self.user_name} sysadmin \
                    password '{macro.COMMON_PASSWD}'; '''
        self.logger.info(sql_cmd)
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd)
        self.logger.info(sql_msg)
        self.assertIn('reset_unique_sql' and '(1 row)',
                      sql_msg, '执行失败:' + text)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG and
                      self.constant.CREATE_ROLE_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        text = '---step3.1:切换用户,创建part表,expect:创建成功 ---'
        self.logger.info(text)
        create_part = f'''set role {self.user_name} \
                    password '{macro.COMMON_PASSWD}';
                    set current_schema to public;
                    drop table if exists {self.table_name_1};
                    create table {self.table_name_1}
                    (
                    p_partkey     bigint not null, 
                    p_name        varchar(55) not null, 
                    p_mfgr        char(25) not null, 
                    p_brand       char(10) not null, 
                    p_type        varchar(25) not null, 
                    p_size        bigint not null, 
                    p_container   char(10) not null, 
                    p_retailprice decimal(15,2) not null, 
                    p_comment     varchar(23) not null
                    )
                    partition by range(p_size)
                    (
                    partition p_size_1 values less than(11),
                    partition p_size_2 values less than(21),
                    partition p_size_3 values less than(31),
                    partition p_size_4 values less than(41),
                    partition p_size_5 values less than(51)
                    );'''
        msg_part = self.sh_primy.execut_db_sql(create_part)
        self.logger.info(msg_part)
        self.assertTrue(msg_part.count(self.constant.SET_SUCCESS_MSG) == 2,
                        '执行失败:' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS, msg_part,
                      '执行失败:' + text)
        text = '---step3.2:切换用户,创建partsupp表,expect:创建成功 ---'
        self.logger.info(text)
        create_partsupp = f'''set role {self.user_name} \
                    password '{macro.COMMON_PASSWD}';
                    set current_schema to public;
                    drop table if exists {self.table_name_2};
                    create table {self.table_name_2}
                    (
                    ps_partkey     bigint not null, 
                    ps_suppkey     bigint not null, 
                    ps_availqty    bigint not null, 
                    ps_supplycost  decimal(15,2)  not null, 
                    ps_comment     varchar(199) not null
                    )
                    partition by range(ps_availqty)
                    (
                    partition ps_availqty_1 values less than(1000),
                    partition ps_availqty_2 values less than(2000),
                    partition ps_availqty_3 values less than(3000),
                    partition ps_availqty_4 values less than(4000),
                    partition ps_availqty_5 values less than(5000),
                    partition ps_availqty_6 values less than(6000),
                    partition ps_availqty_7 values less than(7000),
                    partition ps_availqty_8 values less than(8000),
                    partition ps_availqty_9 values less than(9000),
                    partition ps_availqty_10 values less than(10000)
                    );'''
        msg_partsupp = self.sh_primy.execut_db_sql(create_partsupp)
        self.logger.info(msg_partsupp)
        self.assertTrue(msg_partsupp.count(self.constant.SET_SUCCESS_MSG) == 2,
                        '执行失败:' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS,
                      msg_partsupp, '执行失败:' + text)
        text = '---step3.3:切换用户,创建supplier表,expect:创建成功 ---'
        self.logger.info(text)
        create_supplier = f'''set role {self.user_name} \
                    password '{macro.COMMON_PASSWD}';
                    set current_schema to public;
                    drop table if exists {self.table_name_3};
                    create table {self.table_name_3}
                    (s_suppkey     bigint not null, 
                    s_name        char(25) not null, 
                    s_address     varchar(40) not null, 
                    s_nationkey   int not null, 
                    s_phone       char(15) not null, 
                    s_acctbal     decimal(15,2) not null, 
                    s_comment     varchar(101) not null
                    )
                    partition by range(s_nationkey)
                    (
                    partition s_nationkey_1 values less than(1),
                    partition s_nationkey_2 values less than(2),
                    partition s_nationkey_3 values less than(3),
                    partition s_nationkey_4 values less than(4),
                    partition s_nationkey_5 values less than(5),
                    partition s_nationkey_6 values less than(6),    
                    partition s_nationkey_7 values less than(7),
                    partition s_nationkey_8 values less than(8),
                    partition s_nationkey_9 values less than(9),
                    partition s_nationkey_10 values less than(10),
                    partition s_nationkey_11 values less than(11),
                    partition s_nationkey_12 values less than(12),
                    partition s_nationkey_13 values less than(13),
                    partition s_nationkey_14 values less than(14),
                    partition s_nationkey_15 values less than(15),
                    partition s_nationkey_16 values less than(16),
                    partition s_nationkey_17 values less than(17),
                    partition s_nationkey_18 values less than(18),
                    partition s_nationkey_19 values less than(19),
                    partition s_nationkey_20 values less than(20),
                    partition s_nationkey_21 values less than(21),
                    partition s_nationkey_22 values less than(22),
                    partition s_nationkey_23 values less than(23),
                    partition s_nationkey_24 values less than(24),
                    partition s_nationkey_25 values less than(25));'''
        msg_supplier = self.sh_primy.execut_db_sql(create_supplier)
        self.logger.info(msg_supplier)
        self.assertTrue(msg_partsupp.count(self.constant.SET_SUCCESS_MSG) == 2,
                        '执行失败:' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS,
                      msg_supplier, '执行失败:' + text)

        text = '---step4:导入数据，expect:导入成功---'
        self.logger.info(text)
        mkdir_cmd = f'rm -rf {self.table_path};' \
                    f'mkdir {self.table_path}'
        self.userNode.sh(mkdir_cmd)
        self.common.scp_file(self.userNode, 'part.tbl',
                             self.table_path)
        self.common.scp_file(self.userNode, 'partsupp.tbl',
                             self.table_path)
        self.common.scp_file(self.userNode, 'supplier.tbl',
                             self.table_path)
        part_path = os.path.join(self.table_path, 'part.tbl')
        partsupp_path = os.path.join(self.table_path, 'partsupp.tbl')
        supplier_path = os.path.join(self.table_path, 'supplier.tbl')
        copy_cmd = f'''set role {self.user_name} \
            password '{macro.COMMON_PASSWD}';
            copy {self.table_name_1} from '{part_path}' using delimiters '|'; 
            copy {self.table_name_2} from '{partsupp_path}' \
            using delimiters '|';
            copy {self.table_name_3} from '{supplier_path}' \
            using delimiters '|';
            analyze;'''
        self.logger.info(copy_cmd)
        copy_msg = self.sh_primy.execut_db_sql(copy_cmd)
        self.logger.info(copy_msg)
        copy_assert = copy_msg.count(self.constant.copy_success_msg) == 3
        self.assertTrue(copy_assert, '执行失败:' + text)
        self.assertIn(self.constant.SET_SUCCESS_MSG and
                      self.constant.ANALYZE_SUCCESS_MSG,
                      copy_msg, '执行失败:' + text)

    def test_fullsql(self):
        text = "---step5:设置log_min_duration_statement='24d'," \
               "设置track_stmt_stat_level='l0,l2',执行query," \
               "expect:设置成功,执行query成功 ---"
        self.logger.info(text)
        sql_cmd = f'''set role {self.user_name} \
            password '{macro.COMMON_PASSWD}';
            set log_min_duration_statement='24d';
            show log_min_duration_statement;
            set track_stmt_stat_level='l0,l2';
            show track_stmt_stat_level;
            select pg_sleep(60);
            select
                p_brand,p_type,p_size,
                count(distinct ps_suppkey) as supplier_cnt
            from
                {self.table_name_2}, {self.table_name_1}
            where
                p_partkey = ps_partkey
                and p_brand <> 'Brand#45'
                and p_type not like 'MEDIUM POLISHED%'
                and p_size in (49, 14, 23, 45, 19, 3, 36, 9)
                and ps_suppkey not in (
                    select s_suppkey
                    from {self.table_name_3}
                    where s_comment like '%Customer%Complaints%')
            group by
                p_brand, p_type, p_size
            order by
                supplier_cnt desc, p_brand, p_type, p_size
            limit 100;
            select pg_sleep(100);'''
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd)
        self.logger.info(sql_msg)
        self.assertTrue(sql_msg.count(self.constant.SET_SUCCESS_MSG) == 3,
                        '执行失败:' + text)
        self.assertIn('24d' and 'l0,l2' and '(100 rows)',
                      sql_msg, '执行失败:' + text)

        text = '---step6:查询dbe_perf.get_slow_sql_by_timestamp函数,' \
               'expect:slowsql无记录---'
        self.logger.info(text)
        sql_cmd = f'''select count(*) > 0 
            from dbe_perf.get_global_slow_sql_by_timestamp\
                (now()-interval'15 min',now()) 
            where unique_query_id=(select unique_sql_id 
                from dbe_perf.statement 
                where query like '%count(distinct ps_suppkey) as supplier_cnt%'
                      and user_name = '{self.user_name}');'''
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg,
                                  '?column?', 'f', '(1 row)', flag='1')

        text = '---step7:查询dbe_perf.get_full_sql_by_timestamp函数,' \
               'expect:slowsql有记录---'
        self.logger.info(text)
        sql_cmd = f'''select count(*) > 0 
            from dbe_perf.get_global_full_sql_by_timestamp\
                (now()-interval'15 min',now()) 
            where unique_query_id=(select unique_sql_id 
                from dbe_perf.statement 
                where query like '%count(distinct ps_suppkey) as supplier_cnt%' 
                    and user_name = '{self.user_name}');'''
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.common.equal_sql_mdg(sql_msg,
                                  '?column?', 't', '(1 row)', flag='1')

    def tearDown(self):
        text = '---step8:环境清理，expect:成功---'
        self.logger.info(text)
        text = '---step8.1检查参数，恢复原值:enable_copy_server_files，' \
               'expect:成功---'
        self.logger.info(text)
        check_copy = self.sh_primy.execut_db_sql(self.show_copy_cmd)
        self.logger.info(check_copy)
        current = check_copy.splitlines()[-2].strip()
        if self.copy_default != current:
            result = self.sh_primy.execute_gsguc('reload',
                                                 self.constant.GSGUC_SUCCESS_MSG,
                                                 f'{self.config_copy}={self.copy_default}')
            self.assertTrue(result, '参数恢复失败')
        show_copy_msg = self.sh_primy.execut_db_sql(self.show_copy_cmd)
        self.logger.info(show_copy_msg)

        text = '---step8.2检查参数，恢复原值:instr_unique_sql_count，' \
               'expect:成功---'
        self.logger.info(text)
        check_instr = self.sh_primy.execut_db_sql(self.show_instr_cmd)
        self.logger.info(check_instr)
        current = check_instr.splitlines()[-2].strip()
        if self.instr_default != current:
            result = self.sh_primy.execute_gsguc('reload',
                                                 self.constant.GSGUC_SUCCESS_MSG,
                                                 f'{self.config_instr}={self.instr_default}')
            self.assertTrue(result, '参数恢复失败')
        show_instr_msg = self.sh_primy.execut_db_sql(self.show_instr_cmd)
        self.logger.info(show_instr_msg)

        text = '---step8.3删除表，删除用户,删除文件夹，expect:成功---'
        self.logger.info(text)
        drop_cmd = f'''drop owned by {self.user_name} cascade;
                       drop user {self.user_name};'''
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd)
        self.logger.info(drop_msg)
        rm_cmd = f'rm -rf {self.table_path};'
        self.rootNode.sh(rm_cmd).result()
        self.assertIn(self.constant.DROP_OWNED_SUCCESS and
                      self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败:' + text)
        self.assertFalse(os.path.exists(self.table_path), '执行失败:' + text)
        self.common.equal_sql_mdg(show_copy_msg, self.config_copy,
                                  self.copy_default, '(1 row)', flag='1')
        self.common.equal_sql_mdg(show_instr_msg, self.config_instr,
                                  self.instr_default, '(1 row)', flag='1')
        self.logger.info(
            '---Opengauss_Function_Fullsql_Case0015 finish---')
